package com.kafka.produce;

import com.alibaba.fastjson.JSONObject;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.requestreply.ReplyingKafkaTemplate;
import org.springframework.kafka.requestreply.RequestReplyFuture;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author zhangbowen
 * @date 2019/3/21
 **/
@RestController
public class Produce {
    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
    @Autowired
    private ReplyingKafkaTemplate<String, String, String> replyingKafkaTemplate;

    @GetMapping(value = "publish")
    public void publish() {
        // for (int i=0;i<4;i++) {
        RequestReplyFuture<String, String, String> requestReplyFuture = replyingKafkaTemplate.sendAndReceive(new ProducerRecord<>("test_topic_1", 0, "key", "我是的一条数据"));
        requestReplyFuture.getSendFuture().addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(JSONObject.toJSON(throwable));
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                RecordMetadata metadata = stringStringSendResult.getRecordMetadata();
                System.out.println("Send ok" + metadata);
            }
        });
       /* requestReplyFuture.addCallback(new ListenableFutureCallback<ConsumerRecord<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(JSONObject.toJSON(throwable));
            }

            @Override
            public void onSuccess(ConsumerRecord<String, String> record) {
                System.out.println("收到回复" + record.value());
            }
        });*/
      /*  try {
            Thread.sleep(3000);
        }catch (Exception e){
            e.printStackTrace();
        }*/
        try {
            ConsumerRecord<String, String> consumerRecord = requestReplyFuture.get();
            System.out.println("Return value: " + consumerRecord.value());
        } catch (Exception e) {
            e.printStackTrace();
        }
        //  }
    }

    public void syncPublish() {
        try {
            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test_topic_1", 0, "key", "我是的一条数据0");
            // ListenableFuture<SendResult<String, String>> future1 = kafkaTemplate.send("test_topic_1", 1, "key", "我是的一条数据1");
            // ListenableFuture<SendResult<String, String>> future2 = kafkaTemplate.send("test_topic_1", 2, "key", "我是的一条数据2");
            // ListenableFuture<SendResult<String, String>> future3 = kafkaTemplate.send("test_topic_1", 3, "key", "我是的一条数据3");
            SendResult<String, String> result = future.get(10, TimeUnit.SECONDS);
            //已由服务器确认的记录的元数据
            RecordMetadata metadata = result.getRecordMetadata();
            //要发送给Kafka的键/值对。这包括要将记录发送到的主题名称，可选的分区号以及可选的键和值。
            ProducerRecord record = result.getProducerRecord();
            System.out.println(record.toString());
            System.out.println(metadata.toString());
        } catch (ExecutionException | TimeoutException | InterruptedException e) {
            System.out.println(JSONObject.toJSONString(e));
        }
    }

    //异步发送消息
    public void asyncPublish() {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send("test_topic_1", 0, "key", "我是的一条数据1");
        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println(JSONObject.toJSON(throwable));
            }

            @Override
            public void onSuccess(SendResult<String, String> stringStringSendResult) {
                ProducerRecord record = stringStringSendResult.getProducerRecord();
                RecordMetadata metadata = stringStringSendResult.getRecordMetadata();
                System.out.println(record.toString());
                System.out.println(metadata.toString());
            }
        });
    }
}
